// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: xguo

#include "coin2/exchange/base/order/processor.h"

#include <cmath>
#include <ctgmath>
#include <map>
#include <memory>
#include <string>
#include <utility>
#include <vector>

#include <nlohmann/json.hpp>
#include <sys/syscall.h>
#include <sys/types.h>

#include "absl/container/flat_hash_set.h"
#include "coin2/exchange/base/api_base/fix/fix_trade_client_interface.h"
#include "coin2/exchange/base/log/latency_recorder.h"
#include "coin2/exchange/base/log/latency_recorder_global.h"
#include "coin2/exchange/base/order/fix_trade_client_mgr.h"
#include "coin2/exchange/base/order/grpc_trade_client_mgr.h"
#include "coin2/exchange/base/order/system_critical_error.h"
#include "coin2/exchange/bitmex/api_order/native_private_client.h"
#include "coin2/exchange/kraken/api_order/native_private_client.h"

namespace coin2::exchange::base::order {

namespace {

bool GetRestError(const std::shared_ptr<AsioHttpContext>& http_context, std::string* error) {
  if (!http_context) {
    return false;
  }

  if (http_context->status == AsioHttpContextStatus::kTimeout ||
      http_context->status == AsioHttpContextStatus::kError) {
    nlohmann::json info;
    info["error_code"] = nlohmann::json::parse(ProtoToJsonString(http_context->error_code));
    info["target"] = http_context->target;
    info["request_timestamp"] = http_context->request_timestamp;
    info["response_timestamp"] = http_context->response_timestamp;
    *error = info.dump();
    return true;
  }

  return false;
}

bool IsRawOgLogEnabled(const MarketExchangeApi& mea) {
  using namespace coin2::exchange::base::market;

  std::vector<MarketExchangeApi> mea_list = {
      "Spot.Binance.v1"_mea,
      "Futures.Binance.v1"_mea,
      "Spot.Huobi.v1"_mea,
      "Futures.Huobi.v1"_mea,
      "Spot.Okex.v3"_mea,
      "Futures.Okex.v3"_mea,
      "Futures.Prex.v1"_mea,
      "Futures.Bybit.v2"_mea,
      "Futures.Ftx.v1"_mea,
      "Spot.Ftx.v1"_mea,
      "Futures.Deribit.v2"_mea,
      "Options.Deribit.v2"_mea,
      "Spot.Gateio.v4"_mea,
      "Futures.Gateio.v4"_mea,
      "Spot.Mexc.v3"_mea,
      // "Spot.Lmaxdigital.v1"_mea,
  };

  for (const auto& each : mea_list) {
    if (each.market == mea.market && each.exchange == mea.exchange) {
      return true;
    }
  }
  return false;
}

inline int64_t GenRawMsgId(int64_t msg_id, int64_t msg_type) {
  return (msg_id & 0x00FFFFFFFFFFFFLL) | (msg_type << (64 - 8));
}

}  // namespace

WsSubscriber::WsSubscriber(
    const MarketExchangeApi& mea,
    base::api_base::LiveAsyncExecutor* flow_sub,
    const OrderConnectionConfig& conn_config)
    : BaseRawAsyncSubscriber(flow_sub), mea_(mea) {
  std::ifstream ifs("data/coin2/order/og_ws_info.json");
  nlohmann::json jf = nlohmann::json::parse(ifs);
  if (jf.find(mea_.String()) != jf.end()) {
    auto data = jf[mea_.String()];
    config_.set_ws_host(data["host"].get<std::string>());
    config_.set_ws_host_port(data["port"].get<std::string>());
    config_.set_ws_disable_ssl(!data["use_ssl"].get<bool>());
    config_.MergeFrom(conn_config);
    spec_ = std::make_unique<PrivateWsSubscriberSpec>(
        config_.ws_host(),
        config_.ws_host_port(),
        data["path"].get<std::string>(),
        data["decompress_type"].get<std::string>(),
        data["topic"].get<std::string>(),
        data["ping_period_sec"].get<int>(),
        data["timeout_sec"].get<int>(),
        data["handle_server_ping"].get<bool>(),
        !config_.ws_disable_ssl());
  } else {
    auto empty_spec = PrivateWsSubscriberSpec("", "", "", "", "", 0, 0, false, false);
    spec_ = std::make_unique<PrivateWsSubscriberSpec>(empty_spec);
  }
}

void WsSubscriber::SubscribeAsync() {
  conn_.reset(new base::order::BasePrivateWsConnection(
      *spec_,
      std::make_unique<BasePrivateWsSubscriberPolicy>(mea_, config_, native_symbols()),
      std::bind(&WsSubscriber::SubscribeAsync, this),
      flow_sub_->MainIoc(),
      mea_));
  conn_->SetProcessor(processor_);

  if (CheckBlacklistedMea(mea_)) {
    LOG(WARNING) << "[WsSubscriber] disabled private ws for mea: " << mea_.String();
    return;
  }

  // set requires fields
  conn_->SetListenKey(listen_key_);
  conn_->SetUserId(user_id_);
  conn_->SetBullet(bullet_info_);

  conn_->RunAsync();
}

void BaseOrderProcessor::PreProcessMessage(const RawMessage& msg) {
  CHECK_GT(msg.timestamp, 0);
  msg.mea = mea_;
  msg_id_++;
  if (msg.raw_msg_id == 0) {
    msg.raw_msg_id = GenRawMsgId(msg_id_, msg.type);
  }
  if (logger_) {
    logger_->set_raw_msg_id(msg.raw_msg_id);
  }
  UpdateTimestamp(msg.timestamp);
}

void BaseOrderProcessor::ProcessRawMessage(const RawMessage& msg) {
  PreProcessMessage(msg);
  parser()->UpdateTimestamp(get_ts());
  order_manager()->UpdateTimestamp(get_ts());
  order_manager()->ClearResult();

  executor::TopicRecordPacket packet;
  packet.timestamp = get_ts();
  mutable_upd_mgr()->onBeginPacket(packet);
  std::vector<std::pair<const google::protobuf::Message*, OrderSubTopicId>> proto_msgs;

  // Only 5 types of messages
  // 1. SYSTEM
  // 2. ORDER_EXECUTION
  // 3. WS_MESSAGE
  // 4. REST_MESSAGE
  // 5. QUICKFIX_MESSAGE
  if (msg.type == OrderSubTopicId::SYSTEM) {
    // Move this to parser later.
    ws_reset_num_++;
    if (mea_.exchange == coin::proto::Lmaxdigital && msg.error_code.error_message() == "logout") {
      order_manager()->HandleAutoCancelAllProduct(get_ts());
    }
  } else if (msg.type == OrderSubTopicId::ORDER_EXECUTION) {
    order_manager()->HandleOrderExecution(msg.order_execution);
  } else {
    // REST_MESSAGE, WS_MESSAGE and FIX MESSAGES are handled in parser.
    const auto& result = parser()->ParseRawMessage(msg);
    if (!result.has_error()) {
      if (result.is_account_balance()) {
        // successfully query/subscribe account balance
        // 1. initialize currency balance (no repeatly initialization made inside)
        mutable_upd_mgr()->SetAllZeroBalance();
        // 2. update/publish currency balance if have
        if (result.account_balance().each_balance_size() == 0) {
          // ignore
        } else {
          UpdateAccountBalance(result.account_balance());
          proto_msgs.push_back(
              {&result.account_balance(), OrderSubTopicId::PARSED_ACCOUNT_BALANCE});
        }
      }

      if (result.is_account_position()) {
        // successfully query/subscribe account position
        // 1. initialize product position (no repeatly initialization made inside)
        mutable_upd_mgr()->SetAllZeroPosition();
        // 2. update/publish product position if have
        if (result.account_position().each_position_size() > 0) {
          auto acc_position_in_symcache = result.account_position();
          auto position_not_in_symcache = FilterAccountPosition(&acc_position_in_symcache);
          HandleUndetectedPosition(std::move(position_not_in_symcache));
          if (acc_position_in_symcache.each_position_size() > 0) {
            UpdateAccountPosition(acc_position_in_symcache);
          }
          proto_msgs.push_back(
              {&result.account_position(), OrderSubTopicId::PARSED_ACCOUNT_POSITION});
        }
      }

      if (result.is_transfer()) {
        // successfully query/subscribe currency transfer
        // 1. update/publish currency transfer if have
        if (result.currency_transfer_bundle().each_transfer_size() > 0) {
          UpdateCurrencyTransfer(result.currency_transfer_bundle());
          proto_msgs.push_back(
              {&result.currency_transfer_bundle(), OrderSubTopicId::PARSED_TRANSFER});
        }
      }

      if (result.is_funding_fee() && result.funding_fee_bundle().each_funding_fee_size() > 0) {
        UpdateFundingFee(result.funding_fee_bundle());
        proto_msgs.push_back({&result.funding_fee_bundle(), OrderSubTopicId::PARSED_FUNDING_FEE});
      }

      // web3 specific
      if (result.is_nft_token_info()) {
        UpdateNftTokenInfo(result.nft_token_info());
        proto_msgs.push_back({&result.nft_token_info(), OrderSubTopicId::PARSED_NFT_TOKEN_INFO});
      }

      if (result.is_cancel_response()) {
        order_manager()->HandleCancelResponse(result.cancel_response());
        proto_msgs.push_back({&result.cancel_response(), OrderSubTopicId::PARSED_CANCEL_RESPONSE});
      }

      if (result.is_multi_cancel_response()) {
        order_manager()->HandleCancelBatchResponse(result.multi_cancel_response());
        proto_msgs.push_back(
            {&result.multi_cancel_response(), OrderSubTopicId::PARSED_MULTI_CANCEL_RESPONSE});
      }

      if (result.is_submit_response()) {
        order_manager()->HandleSubmitResponse(result.submit_response());
        proto_msgs.push_back({&result.submit_response(), OrderSubTopicId::PARSED_SUBMIT_RESPONSE});
      }

      if (result.is_multi_submit_response()) {
        order_manager()->HandleSubmitBatchResponse(result.multi_submit_response());
        proto_msgs.push_back(
            {&result.multi_submit_response(), OrderSubTopicId::PARSED_MULTI_SUBMIT_RESPONSE});
      }

      // treat conversion reponse as query history conversion response
      if (result.is_conversion() && result.conversion_list().each_conversion_size() > 0) {
        order_manager()->HandleConversionList(result.conversion_list());
        proto_msgs.push_back({&result.conversion_list(), OrderSubTopicId::PARSED_CONVERSION});
      }
      // Handle fill before order, because traders prefer to recv fill before order is dead.
      if (result.is_fill() && result.fill_list().each_fill_size() > 0) {
        order_manager()->HandleFillInfoList(result.fill_list());
        proto_msgs.push_back({&result.fill_list(), OrderSubTopicId::PARSED_ORDER_FILL});
      }

      if (result.is_order() && result.order_list().each_order_size() > 0) {
        order_manager()->HandleOrderInfoList(result.order_list());
        proto_msgs.push_back({&result.order_list(), OrderSubTopicId::PARSED_ORDER_STATUS});
        // Fill is handled twice, assume all fill has fill_id.
        // First time some fill may not found its corresponding order (order is not created
        // yet!)
        if (result.is_fill() && result.fill_list().each_fill_size() > 0) {
          order_manager()->HandleFillInfoList(result.fill_list());
        }
      }
    } else {
      LOG(ERROR) << "Parsing error: " << msg.DebugString();
    }
  }
  mutable_upd_mgr()->onEndPacket(packet);

  auto ts = get_ts();
  if (last_check_timeout_order_ts_ + 1'000'000'000LL <= ts) {
    last_check_timeout_order_ts_ = ts;
    order_manager()->RemoveTimeoutPendingOrders(ts);
    order_manager()->ManageRisk(ts);
    risk_undetected_position_->ManageRisk(ts);
  }
  // hack to make it work - check zombie order
  if (mea_.exchange == coin::proto::Prex) {
    static const std::string event_id = mea_.market == coin::proto::Spot
                                            ? "prex_spot/query_order_by_ids"
                                            : "prex_futures/query_order_by_ids";
    const int64_t max_age = 30'000'000'000LL;
    std::stringstream exchange_order_ids;
    std::stringstream mixin_order_ids;
    bool has_zombie_order = false;
    int num_orders = 0;
    for (const auto poid : order_manager()->GetZombieOrders(ts, max_age)) {
      const auto* oc = order_manager()->FindByProcOrderId(poid);
      if (oc->has_exchange_order_id()) {
        num_orders++;
        if (has_zombie_order) {
          exchange_order_ids << "," << oc->exchange_order_id();
          mixin_order_ids << "," << poid << "@" << oc->exchange_order_id();
        } else {
          exchange_order_ids << oc->exchange_order_id();
          mixin_order_ids << poid << "@" << oc->exchange_order_id();
        }
        has_zombie_order = true;
      } else {
        DLOG(ERROR) << "Zombie order submitted but not accepted! " << oc->ToProto().DebugString();
      }

      if (num_orders % 50 == 0 && has_zombie_order) {
        auto info = fmt::format("{}|{}", exchange_order_ids.str(), mixin_order_ids.str());
        mb_->PublishEvent(event_id, &info);
        has_zombie_order = false;
        exchange_order_ids.str(std::string());
        mixin_order_ids.str(std::string());
      }
    }

    if (has_zombie_order) {
      auto info = fmt::format("{}|{}", exchange_order_ids.str(), mixin_order_ids.str());
      mb_->PublishEvent(event_id, &info);
      has_zombie_order = false;
    }
  } else if (mea_.exchange == coin::proto::Lmaxdigital) {
    // skip checking lmaxdigital
  } else {
    static int64_t max_order_idle_age = [this]() -> int64_t {
      int64_t default_age_sec = 60;
      int64_t dynamic_age_sec = rest_subscriber_->GetConfig().query_open_orders_period_sec() * 2;
      return std::max(default_age_sec, dynamic_age_sec) * 1'000'000'000LL;
    }();
    if (last_check_zombie_order_ts_ + 1'000'000'000LL <= ts) {
      last_check_zombie_order_ts_ = ts;
      order_manager()->RemoveZombieOrders(ts, max_order_idle_age);
    }
  }

  // write order context debug info if there is any.
  const auto& result = order_manager()->result();
  if (result.has_debug_info()) {
    logger_->WriteOrderEvent(result.order_event, true, get_ts());
  }

  // write raw and parsed message.
  if (enable_raw_og_log_) {
    order_packet_logger_->set_raw_msg_id(msg.raw_msg_id);
    order_packet_logger_->WritePrivateExchangeMessage(msg.SerializeToString(), true, get_ts());

    if (store_parsed_message_) {
      for (const auto& pmsg : proto_msgs) {
        RawMessage parsed_msg;
        parsed_msg.timestamp = get_ts();
        parsed_msg.proto_msg = pmsg.first;
        parsed_msg.type = pmsg.second;
        order_packet_logger_->WritePrivateExchangeMessage(
            parsed_msg.SerializeToString(),
            true,
            get_ts());
      }
    }
  }
}

void BaseOrderProcessor::HandleHttpError(const std::shared_ptr<AsioHttpContext>& http_context) {
  if (!http_context) {
    return;
  }
  std::string rest_error;
  if (!GetRestError(http_context, &rest_error)) {
    return;
  }
  OrderEvent order_event;
  order_event.set_type(OrderEvent::DEBUG_INFO);
  order_event.set_tag(rest_error);
  logger_->WriteOrderEvent(order_event, true, get_ts());

  if (http_context->status == AsioHttpContextStatus::kTimeout) {
    executor::TopicRecordPacket packet;
    packet.timestamp = get_ts();
    mutable_upd_mgr()->onBeginPacket(packet);
    mutable_upd_mgr()->SetOrderTimeout();
    callback_->onOrderSubsystemError(mutable_upd_mgr()->update());
    mutable_upd_mgr()->onEndPacket(packet);

    if (latency_recorder_) {
      latency_recorder_->SetTimeoutMark(logger_->request(), rest_error);
    }
  }

  // If fd exhausted, restart the strategy! Should never happen!!!
  CHECK_NE(http_context->error_code.error_code(), boost::system::errc::device_or_resource_busy)
      << http_context->ToProto().DebugString();
}

void BaseOrderProcessor::ProcessUndetectPosition(
    ProductPosition* position,
    std::vector<ProductPosition>* position_not_in_symcache) {
  const auto& native_symbol = position->native_symbol();

  if (position->long_position() == 0 && position->short_position() == 0 &&
      position->net_position() == 0) {
    return;
  }
  if (native_symbol_to_product_map_.count(native_symbol) == 0) {
    auto product = CreateProductFromNativeStringNoThrow(mea_, native_symbol, get_ts());
    if (!product) {
      LOG(ERROR) << "Get product fail! mea:" << mea_.String() << " native symbol:" << native_symbol;
      return;
    }
    native_symbol_to_product_map_.emplace(native_symbol, std::move(product));
  }
  auto& product = native_symbol_to_product_map_[native_symbol];
  position->set_symbol(product->absolute_norm());
  position_not_in_symcache->emplace_back(std::move(*position));
}

std::vector<ProductPosition> BaseOrderProcessor::FilterAccountPosition(
    AccountPosition* acc_position) {
  auto* position = acc_position->mutable_each_position();
  std::vector<ProductPosition> position_not_in_symcache;

  for (auto it = position->begin(); it != position->end();) {
    if (symcache_->has_product_native(it->native_symbol())) {
      auto* product = symcache_->GetProductByNative(it->native_symbol());
      it->set_symbol(product->symbol());
      ++it;
    } else {
      ProcessUndetectPosition(&(*it), &position_not_in_symcache);
      it = position->erase(it);
    }
  }

  return position_not_in_symcache;
}

void BaseOrderProcessor::HandleUndetectedPosition(
    std::vector<ProductPosition>&& undetected_position) {
  if (!risk_undetected_position_) {
    return;
  }
  risk_undetected_position_->HandleUndetectedPosition(std::move(undetected_position));
}

void BaseOrderProcessor::HandleOrderEvent(const OrderEvent& order_event) {
  if (order_event.type() == OrderEvent::ORDER_REJECTED) {
    ++info_.order_reject_num;
  }
  if (order_event.type() == OrderEvent::CANCEL_REJECTED) {
    ++info_.cancel_reject_num;
  }
  if (order_event.type() == OrderEvent::ORDER_ACCEPTED) {
    ++info_.order_accept_num;
  }

  if (order_event.type() == OrderEvent::CONVERSION) {
    const std::string& native_symbol = order_event.native_symbol();
    mutable_upd_mgr()->GetProductOrderInfoMutable(native_symbol)->set_event(&order_event);
    mutable_upd_mgr()->PublishAccountOrder(order_event);
  } else {
    const std::string& native_symbol = order_event.native_symbol();
    const int64_t proc_order_id = order_event.proc_order_id();
    const auto* ocm = mutable_upd_mgr()->order_manager();
    const auto* oc = ocm->FindByProcOrderId(proc_order_id);
    CHECK(ocm);
    CHECK(reinterpret_cast<uint64_t>(oc) > 0xFFFF) << oc;

    if (order_event.publish_live()) {
      if (recent_active_symbol_) {
        recent_active_symbol_->AddActiveEvent(get_ts(), native_symbol);
      }
      mutable_upd_mgr()
          ->GetProductOrderInfoMutable(native_symbol)
          ->UpdateOrderState(order_event, oc);
      mutable_upd_mgr()->PublishAccountOrder(order_event);
    }

    if (order_event.fill_source() != coin::proto::FillSource::IMPLIED_FROM_DEAD) {
      logger_->WriteOrderEvent(order_event, true, get_ts());
      order_rate_tracker_->HandleOrderEvent(order_event, oc, get_ts());
    }
  }
}

void BaseOrderProcessor::HandleRiskEvent(const RiskEvent& risk_event) {
  if (logger_) {
    logger_->WriteRiskEvent(risk_event, true, get_ts());
  }
}

BaseOrderGateway::BaseOrderGateway() : og_config_(OrderGatewayConfig()), mea_(MarketExchangeApi()) {
  InitWithMea(mea_);
}

BaseOrderGateway::BaseOrderGateway(
    const MarketExchangeApi& mea,
    const OrderGatewayConfig& og_config,
    executor::LiveAsyncExecutor* exec,
    NativePrivateClient* rest_client,
    base::api_base::NativeHttpAsioDriver* http_driver,
    order_util::OrderLogger* order_logger)
    : og_config_(og_config),
      flow_sub_(exec),
      order_logger_(order_logger),
      latency_recorder_(nullptr),
      http_driver_(http_driver),
      rest_client_(rest_client),
      fix_client_{nullptr},
      mea_{mea} {
  InitWithMea(mea_);
}

void BaseOrderGateway::InitWithMea(const MarketExchangeApi& mea) {
  if (mea.market == MarketType::Futures) {
    is_futures_ = true;
  } else {
    is_futures_ = false;
  }

  if (mea.api.find("swap") != std::string::npos) {
    is_swap_ = true;
  }
  need_exchange_id_to_cancel_ = true;
  is_supporting_ioc_ = false;
  is_supporting_post_only_ = true;
  is_supporting_multi_cancel_order_ = false;
  is_supporting_multi_submit_order_ = false;
  std::ifstream ifs("data/coin2/order/og_info.json");
  nlohmann::json jf = nlohmann::json::parse(ifs);
  if (jf.find(mea.String()) != jf.end()) {
    auto& data = jf[mea.String()];
    if (data.find("need_exchange_id_to_cancel") != data.end()) {
      need_exchange_id_to_cancel_ = data["need_exchange_id_to_cancel"];
    }
    if (data.find("is_supporting_ioc") != data.end()) {
      is_supporting_ioc_ = data["is_supporting_ioc"];
    }
    if (data.find("is_supporting_post_only") != data.end()) {
      is_supporting_post_only_ = data["is_supporting_post_only"];
    }
    if (data.find("is_supporting_multi_cancel_order") != data.end()) {
      is_supporting_multi_cancel_order_ = data["is_supporting_multi_cancel_order"];
    }
    if (data.find("is_supporting_multi_submit_order") != data.end()) {
      is_supporting_multi_submit_order_ = data["is_supporting_multi_submit_order"];
    }
    LOG(INFO) << fmt::format(
        "{}, need_exchange_id_to_cancel {}, is_supporting_ioc {}, is_supporting_post_only {}, "
        "is_supporting_multi_cancel_order {} is_supporting_multi_submit_order {}",
        mea.String(),
        need_exchange_id_to_cancel_,
        is_supporting_ioc_,
        is_supporting_post_only_,
        is_supporting_multi_cancel_order_,
        is_supporting_multi_submit_order_);
  }
}

void BaseOrderGateway::HandleResponse(
    std::vector<const OrderContext*> oc_list,
    std::shared_ptr<AsioHttpContext> http_context) {
  if (http_context->status == AsioHttpContextStatus::kRequested) {
    return;
  }
  GetRateLimitInfo(http_context);
  CHECK(EqualToAnyOf(
      http_context->status,
      AsioHttpContextStatus::kResponded,
      AsioHttpContextStatus::kTimeout,
      AsioHttpContextStatus::kError));

  BaseParser::RawMessage msg;
  msg.type = OrderSubTopicId::REST_MESSAGE;
  msg.http_context = http_context;
  msg.timestamp = http_context->response_timestamp;
  // TODO(daniel): there is no oc for converstion. it's too specify here??
  msg.oc_list = oc_list;
  msg.oc = msg.oc_list.size() == 0 ? nullptr : const_cast<OrderContext*>(msg.oc_list[0]);
  processor_->ProcessRawMessage(msg);
  // write rest query error (read/write/timeout etc)
  processor_->HandleHttpError(http_context);
#ifdef ENABLE_LATENCY_PROFILE_DEBUG_MARK
  auto latency_context_id = msg.http_context->context_id;
  RemoveContextIdFromRegisteredOGSet(latency_context_id);
#endif
  // add latency mark
  if (msg.http_context->status != AsioHttpContextStatus::kResponded) {
    return;
  }
  switch (msg.http_context->query_type) {
    case OrderSubTopicId::SUBMIT_RESPONSE: {
      if (latency_recorder_) {
#ifdef ENABLE_LATENCY_PROFILE_DEBUG_MARK
        LatencyRecorderGlobal::SetLatencyProfileMark(
            order_logger_->request(),
            "OG_HTTP_REQUEST_SENT",
            msg.http_context->request_timestamp,
            &latency_context_id,
            false);
        LatencyRecorderGlobal::SetLatencyProfileMark(
            order_logger_->request(),
            "OG_HTTP_REQUEST_RECV",
            msg.http_context->response_timestamp,
            &latency_context_id,
            false);
        LatencyRecorderGlobal::SetLatencyProfileMark(
            order_logger_->request(),
            "AFTER_OG_ORDER_SUBMIT",
            0,
            &latency_context_id,
            true);
#endif
        latency_recorder_->SetAfterSubmitMark(msg.oc, order_logger_->request());
      }
      break;
    }
    case OrderSubTopicId::AMEND_RESPONSE: {
      if (latency_recorder_) {
#ifdef ENABLE_LATENCY_PROFILE_DEBUG_MARK
        LatencyRecorderGlobal::SetLatencyProfileMark(
            order_logger_->request(),
            "OG_HTTP_REQUEST_SENT",
            msg.http_context->request_timestamp,
            &latency_context_id,
            false);
        LatencyRecorderGlobal::SetLatencyProfileMark(
            order_logger_->request(),
            "OG_HTTP_REQUEST_RECV",
            msg.http_context->response_timestamp,
            &latency_context_id,
            false);
        LatencyRecorderGlobal::SetLatencyProfileMark(
            order_logger_->request(),
            "AFTER_OG_ORDER_AMEND",
            0,
            &latency_context_id,
            true);
#endif
        latency_recorder_->SetAfterAmendMark(msg.oc, order_logger_->request());
      }
      break;
    }
    case OrderSubTopicId::CANCEL_RESPONSE: {
      if (latency_recorder_) {
#ifdef ENABLE_LATENCY_PROFILE_DEBUG_MARK
        LatencyRecorderGlobal::SetLatencyProfileMark(
            order_logger_->request(),
            "OG_HTTP_REQUEST_SENT",
            msg.http_context->request_timestamp,
            &latency_context_id,
            false);
        LatencyRecorderGlobal::SetLatencyProfileMark(
            order_logger_->request(),
            "OG_HTTP_REQUEST_RECV",
            msg.http_context->response_timestamp,
            &latency_context_id,
            false);
        LatencyRecorderGlobal::SetLatencyProfileMark(
            order_logger_->request(),
            "AFTER_OG_ORDER_CANCEL",
            0,
            &latency_context_id,
            true);
#endif
        latency_recorder_->SetAfterCancelMark(msg.oc, order_logger_->request());
      }
      break;
    }
    default:
      break;
  }
}

void BaseOrderGateway::HandleGrpcResponse(std::shared_ptr<GrpcContext> context) {
  boost::asio::post(*flow_sub_->MainIoc(), [context, processor = processor_]() {
    BaseParser::RawMessage raw_msg;
    raw_msg.type = OrderSubTopicId::GRPC_MESSAGE;
    raw_msg.grpc_context = context;
    raw_msg.timestamp = GetCurrentTimestamp();
    processor->ProcessRawMessage(raw_msg);
  });
}

bool BaseOrderGateway::IsAcceptableValue(const OrderSpec& order_spec) const {
  if (is_futures_ && order_spec.reduce_only) {
    return true;
  }
  // order_spec.product is native_symbol
  auto product_info = symcache_->GetProductInfoByNative(order_spec.product);

  double min_amt = product_info->min_amount();
  if (std::isnan(min_amt)) {
    min_amt = 0.0;
  }

  double min_qty = product_info->min_qty();
  if (std::isnan(min_qty)) {
    min_qty = 0.0;
  }
  return (order_spec.price * order_spec.qty >= min_amt) && (order_spec.qty >= min_qty);
}

void BaseOrderGateway::PrepareSubmitOrderExecutionProto(const OrderSpec& order_spec) {
  // 1. order_spec with proc_order_id
  result_.order_execution.set_type(OrderExecutionProto::SUBMIT_ORDER);

  CHECK_GT(order_spec.proc_order_id, 0);
  result_.order_execution.set_proc_order_id(order_spec.proc_order_id);
  *result_.order_execution.mutable_order_spec() = order_spec.ToProto();
#ifdef ENABLE_LATENCY_PROFILE_DEBUG_MARK
  result_.timestamp = GetCurrentTimestamp();
#else
  result_.timestamp = order_spec.order_created_time;
#endif
}

void BaseOrderGateway::PrepareCancelOrderExecutionProto(int64_t proc_order_id) {
  // 1. cancel only need proc_order_id
  CHECK_GT(proc_order_id, 0);
  result_.order_execution.set_type(OrderExecutionProto::CANCEL_ORDER);
  result_.order_execution.set_proc_order_id(proc_order_id);
  result_.timestamp = GetCurrentTimestamp();
}

void BaseOrderGateway::PrepareCancelBatchExecutionProto(
    const std::vector<int64_t>& proc_order_ids) {
  result_.order_execution.clear_proc_order_ids();

  for (const auto& id : proc_order_ids) {
    CHECK_GT(id, 0);
    result_.order_execution.add_proc_order_ids(id);
  }

  result_.order_execution.set_type(OrderExecutionProto::CANCEL_BATCH_ORDERS);
  result_.timestamp = GetCurrentTimestamp();
}

void BaseOrderGateway::PrepareAmendOrderExecutionProto(
    int64_t proc_order_id,
    int64_t orig_proc_order_id,
    double amend_price,
    double amend_qty,
    const OrderSpec& amend_order_spec) {
  // proc_order_id == amend_order_spec.proc_order_id
  // orig_proc_order_id == amend_order_spec.orig_proc_order_id
  CHECK_GT(proc_order_id, 0);
  CHECK_GT(amend_order_spec.proc_order_id, 0);
  result_.order_execution.set_type(OrderExecutionProto::AMEND_ORDER);
  result_.order_execution.set_proc_order_id(proc_order_id);
  result_.order_execution.set_orig_proc_order_id(orig_proc_order_id);
  result_.order_execution.set_amend_price(amend_price);
  result_.order_execution.set_amend_qty(amend_qty);
#ifdef ENABLE_LATENCY_PROFILE_DEBUG_MARK
  result_.timestamp = GetCurrentTimestamp();
#else
  result_.timestamp = amend_order_spec.order_created_time;
#endif
  *result_.order_execution.mutable_order_spec() = amend_order_spec.ToProto();
}

void BaseOrderGateway::PrepareCancelAllExecutionProto(const std::string& native_symbol) {
  result_.order_execution.set_type(OrderExecutionProto::CANCEL_ALL_ORDERS);
  result_.timestamp = GetCurrentTimestamp();
}

void BaseOrderGateway::PrepareSubmitBatchExecutionProto(
    const std::deque<OrderSpec>& order_spec_deque) {
  result_.order_execution.clear_order_specs();

  for (const auto& order_spec : order_spec_deque) {
    // 1. order_spec with proc_order_id
    CHECK_GT(order_spec.proc_order_id, 0);

    *result_.order_execution.add_order_specs() = order_spec.ToProto();
  }

  result_.order_execution.set_type(OrderExecutionProto::SUBMIT_BATCH_ORDERS);
  result_.timestamp = GetCurrentTimestamp();
}

void BaseOrderGateway::PrepareSubmitBatchExecutionProto(
    const std::deque<const OrderSpec*>& order_spec_deque) {
  result_.order_execution.clear_order_specs();

  for (const auto* order_spec : order_spec_deque) {
    // 1. order_spec with proc_order_id
    CHECK_GT(order_spec->proc_order_id, 0);

    *result_.order_execution.add_order_specs() = order_spec->ToProto();
  }

  result_.order_execution.set_type(OrderExecutionProto::SUBMIT_BATCH_ORDERS);
  result_.timestamp = GetCurrentTimestamp();
}

void BaseOrderGateway::SetOrderExecutionError(const std::string& err_msg, int error_code) {
  result_.order_execution.mutable_error_code()->set_category("ORDER_EXECUTION");
  result_.order_execution.mutable_error_code()->set_error_code(error_code);
  result_.order_execution.mutable_error_code()->set_error_message(err_msg);
}

void BaseOrderGateway::SetOrderExecutionError(const std::string& err_msg) {
  SetOrderExecutionError(err_msg, coin::proto::OrderErrorCode::ORDER_EXECUTION_ERROR);
}

std::optional<int64_t> BaseOrderGateway::SubmitOrder(const OrderSpec& order_spec) {
  int64_t proc_order_id = order_manager().AllocateProcOrderId();
  int64_t latency_context_id = 0;
#ifdef ENABLE_LATENCY_PROFILE_DEBUG_MARK
  latency_context_id = GetCurrContextIdWithNoRegisteredOG();
  LatencyRecorderGlobal::SetLatencyProfileMark(
      order_logger_->request(),
      "BEFORE_OG_ORDER_SUBMIT",
      0,
      &latency_context_id,
      false);
#endif

  ClearResult();

  order_spec.proc_order_id = proc_order_id;
  // deal with no_support IOC exchanges
  if ((order_spec.duration == coin::proto::OrderDuration::FOK_ORDER ||
       order_spec.duration == coin::proto::OrderDuration::IOC_ORDER) &&
      !is_supporting_ioc_) {
    order_spec.ResetDuration();
  }
  PrepareSubmitOrderExecutionProto(order_spec);
  int64_t timestamp = result_.timestamp;

  if (!IsAcceptableValue(order_spec)) {
    LOG(ERROR) << "Order not accepted! " << order_spec.DebugString();
    SetOrderExecutionError("ORDER_SPEC_NOT_ACCEPTED");
    processor_->ProcessRawMessage(result_);
    return {};
  }
  // api limit checking. TODO(daniel): can we leave this checking to native client?
  if (order_spec.exchange == ExchangeType::Kraken) {
    auto rest_client = dynamic_cast<KrakenNativePrivateClient*>(rest_client_);
    if (!rest_client->HasEnoughRateLimit(timestamp, ActionType::PLACE, order_spec.product)) {
      LOG(ERROR) << "Hit rate limit. SubmitOrder failed";
      SetOrderExecutionError("SUBMIT_API_RATE_LIMIT_ERROR");
      processor_->ProcessRawMessage(result_);
      return {};
    }
  } else if (order_spec.exchange == ExchangeType::Bitmex) {
    auto rest_client = dynamic_cast<BitmexNativePrivateClient*>(rest_client_);
    rest_client->SetMakerOrder(order_spec.post_only);
    if (!rest_client_->HasEnoughRateLimit(timestamp, ActionType::PLACE)) {
      LOG(ERROR) << "Hit rate limit. SubmitOrder failed";
      SetOrderExecutionError("SUBMIT_API_RATE_LIMIT_ERROR");
      processor_->ProcessRawMessage(result_);
      return {};
    }
  } else if (order_spec.exchange == ExchangeType::Bybit) {
    // leave this checking to native client
  } else if (
      !(fix_client_ && fix_client_->IsAvailable()) &&
      !rest_client_->HasEnoughRateLimit(timestamp, ActionType::PLACE)) {
    LOG(ERROR) << "Hit rate limit. SubmitOrder failed";

    SetOrderExecutionError("SUBMIT_API_RATE_LIMIT_ERROR");
    processor_->ProcessRawMessage(result_);
    return {};
  }

  if (!processor_->order_rate_tracker()->HasEnoughRateLimit(timestamp)) {
    LOG(ERROR) << "Hit rate limit. SubmitOrder failed";

    SetOrderExecutionError("SUBMIT_HUOBI_API_RATE_LIMIT_ERROR");
    processor_->ProcessRawMessage(result_);
    return {};
  }

  OrderExecution order_exe;
  order_exe.order_spec = &order_spec;
  order_exe.proc_order_id = proc_order_id;

  if (grpc_client_) {
    processor_->ProcessRawMessage(result_);
    auto* oc = order_manager().FindByProcOrderId(proc_order_id);
    CHECK(oc);
    order_exe.oc = const_cast<OrderContext*>(oc);
    grpc_client_->PlaceOrder(order_exe);
  } else if (fix_client_ && fix_client_->IsAvailable()) {
    fix_client_->PlaceOrder(order_exe);
    processor_->ProcessRawMessage(result_);
  } else {
    // If submit will be blocked, return immediately.
    if (http_driver_->is_blocking()) {
      SetOrderExecutionError("CONNECTION_POOL_FULL");
      processor_->ProcessRawMessage(result_);
      return {};
    }

    auto request = rest_client_->PlaceOrderNorm(order_exe);
    if (!request) {
      SetOrderExecutionError("API_RATE_LIMIT_ERROR");
      processor_->ProcessRawMessage(result_);
      return {};
    }
    processor_->ProcessRawMessage(result_);

    auto* oc = order_manager().FindByProcOrderId(proc_order_id);
    CHECK(oc);
    std::vector<const OrderContext*> oc_list;
    oc_list.emplace_back(oc);
    request->context_id = latency_context_id;
    request->handler =
        std::bind(&BaseOrderGateway::HandleResponse, this, oc_list, std::placeholders::_1);
    http_driver_->PostAsyncContext(request);
    mutable_order_manager()->PostValidateOrderContext(oc);
  }

  if (latency_recorder_) {
    auto* oc = order_manager().FindByProcOrderId(proc_order_id);
    CHECK(oc);
    latency_recorder_->SetBeforeSubmitMark(oc, order_logger_->request());
  }
  return proc_order_id;
}

bool BaseOrderGateway::CancelOrder(int64_t proc_order_id) {
  int64_t latency_context_id = 0;
#ifdef ENABLE_LATENCY_PROFILE_DEBUG_MARK
  latency_context_id = GetCurrContextIdWithNoRegisteredOG();
  LatencyRecorderGlobal::SetLatencyProfileMark(
      order_logger_->request(),
      "BEFORE_OG_ORDER_CANCEL",
      0,
      &latency_context_id,
      false);
#endif

  ClearResult();

  PrepareCancelOrderExecutionProto(proc_order_id);
  int64_t timestamp = result_.timestamp;

  // check order spec/state that if cancel order is allowed or not
  auto* oc = order_manager().FindByProcOrderId(proc_order_id);
  if (oc == nullptr) {
    // unexpected. should investigate why a non-exist order is trying be canceled
    LOG(ERROR) << "Order not found! " << proc_order_id;
    LOG(INFO) << order_manager().DebugString();

    SetOrderExecutionError("ORDER_NOT_FOUND", coin::proto::OrderErrorCode::ORDER_NOT_FOUND);
    processor_->ProcessRawMessage(result_);
    return false;
  }
  if (oc->is_cancel_pending()) {
    // ignored. that means there no cancel response so far
    if (timestamp - oc->cancel_created_time() > 5e9) {
      LOG(INFO) << "No cancel response for order " << proc_order_id << ", "
                << oc->exchange_order_id();
      // add to cancel_timeout list
      result_.order_execution.add_cancel_timeout_proc_order_ids(oc->proc_order_id());
      SetOrderExecutionError("CANCEL_TIMEOUT");
      processor_->ProcessRawMessage(result_);
    } else {
      LOG(INFO) << "cancel sent already. " << proc_order_id << ", " << oc->exchange_order_id();
    }
    return false;
  }
  if (coin::proto::OrderDuration::IOC_ORDER == oc->order_spec().duration ||
      coin::proto::OrderDuration::FOK_ORDER == oc->order_spec().duration) {
    // ignored. that means IOC/FOK order no submit response so far
    LOG(INFO) << "Not allow cancel IOC/FOK order: " << proc_order_id << ", "
              << oc->exchange_order_id();
    SetOrderExecutionError("CANCEL_IOC_FOK_ORDER");
    processor_->ProcessRawMessage(result_);
    return false;
  }
  if (oc->IsCancelConfirmed() || oc->IsFullyFilled()) {
    // unexpected. should investigate why a dead order is trying to be canceled
    LOG(INFO) << "Ignore cancel dead order: " << proc_order_id << " status: " << oc->status();
    SetOrderExecutionError("ORDER_DEAD_ALREADY");
    processor_->ProcessRawMessage(result_);
    return false;
  }

  // api limit checking. TODO(daniel): can we leave this checking to native client?
  if (oc != nullptr && oc->order_spec().exchange == ExchangeType::Kraken) {
    auto rest_client = dynamic_cast<KrakenNativePrivateClient*>(rest_client_);
    if (!rest_client->HasEnoughRateLimit(timestamp, ActionType::CANCEL, oc->native_product())) {
      LOG(ERROR) << "Hit rate limit. CancelOrder failed";

      SetOrderExecutionError("CANCEL_API_RATE_LIMIT_ERROR");
      processor_->ProcessRawMessage(result_);
      return false;
    }
  } else if (oc != nullptr && oc->order_spec().exchange == ExchangeType::Bybit) {
    // leave this checking to native client
  } else if (
      !(fix_client_ && fix_client_->IsAvailable()) &&
      !rest_client_->HasEnoughRateLimit(timestamp, ActionType::CANCEL)) {
    LOG(ERROR) << "Hit rate limit. CancelOrder failed";

    SetOrderExecutionError("CANCEL_API_RATE_LIMIT_ERROR");
    processor_->ProcessRawMessage(result_);
    return false;
  }
  if (!processor_->order_rate_tracker()->HasEnoughRateLimit(timestamp)) {
    LOG(ERROR) << "Hit rate limit. CancelOrder failed";

    SetOrderExecutionError("CANCEL_API_RATE_LIMIT_ERROR");
    processor_->ProcessRawMessage(result_);
    return false;
  }

  OrderExecution order_exe;
  order_exe.proc_order_id = proc_order_id;
  order_exe.oc = const_cast<OrderContext*>(oc);
  if (fix_client_ && fix_client_->IsAvailable()) {
    fix_client_->CancelOrder(order_exe);
  } else {
    if (http_driver_->is_blocking()) {
      SetOrderExecutionError("CONNECTION_POOL_FULL");
      processor_->ProcessRawMessage(result_);
      return false;
    }

    auto request = rest_client_->CancelOrderNorm(order_exe);
    if (!request) {
      SetOrderExecutionError("CANCEL_API_LIMIT_ERROR");
      processor_->ProcessRawMessage(result_);
      return false;
    }
    std::vector<const OrderContext*> oc_list;
    oc_list.emplace_back(oc);
    request->context_id = latency_context_id;
    request->handler =
        std::bind(&BaseOrderGateway::HandleResponse, this, oc_list, std::placeholders::_1);
    http_driver_->PostAsyncContext(request);
  }

  if (latency_recorder_) {
    latency_recorder_->SetBeforeCancelMark(oc, order_logger_->request());
  }

  processor_->ProcessRawMessage(result_);

  return true;
}

int BaseOrderGateway::CancelBatchOrders(const BatchCancelSpec& batch_cancel_spec) {
  CHECK(IsSupportMultiCancelOrder())
      << "[" << __func__ << "] unsupport cancel batch order. mea: " << mea_.String();

  ClearResult();

  if (batch_cancel_spec.Empty()) {
    LOG(ERROR) << "[" << __func__ << "] ignored. no order need cancel";
    return 0;
  }

  auto proc_order_ids = batch_cancel_spec.GetProcOrderIds();  // need a copy

  std::unordered_map<std::string, std::vector<int64_t>> fail_proc_order_id_map;

  std::deque<const OrderContext*> acceptable_order_context;

  auto timestamp = GetCurrentTimestamp();

  for (const auto& proc_order_id : proc_order_ids) {
    // check order spec/state that if cancel order is allowed or not
    const auto* oc = order_manager().FindByProcOrderId(proc_order_id);
    if (oc == nullptr) {
      // unexpected. should investigate why a non-exist order is trying be canceled
      LOG(ERROR) << "Order not found! " << proc_order_id;
      LOG(INFO) << order_manager().DebugString();

      fail_proc_order_id_map["ORDER_NOT_FOUND"].emplace_back(proc_order_id);
      continue;
    }

    if (oc->is_cancel_pending()) {
      // ignored. that means there no cancel response so far
      if (timestamp - oc->cancel_created_time() > 5e9) {
        LOG(INFO) << "No cancel response for order " << proc_order_id << ", "
                  << oc->exchange_order_id();
        fail_proc_order_id_map["CANCEL_TIMEOUT"].emplace_back(proc_order_id);
      } else {
        LOG(INFO) << "cancel sent already. " << proc_order_id << ", " << oc->exchange_order_id();
      }
      continue;
    }
    if (coin::proto::OrderDuration::IOC_ORDER == oc->order_spec().duration ||
        coin::proto::OrderDuration::FOK_ORDER == oc->order_spec().duration) {
      // ignored. that means IOC/FOK order no submit response so far
      LOG(INFO) << "Not allow cancel IOC/FOK order: " << proc_order_id << ", "
                << oc->exchange_order_id();
      fail_proc_order_id_map["CANCEL_IOC_FOK_ORDER"].emplace_back(proc_order_id);
      continue;
    }

    if (oc->IsCancelConfirmed() || oc->IsFullyFilled()) {
      // unexpected. should investigate why a dead order is trying to be canceled
      LOG(INFO) << "Ignore cancel dead order: " << proc_order_id << " status: " << oc->status();
      fail_proc_order_id_map["ORDER_DEAD_ALREADY"].emplace_back(proc_order_id);
      continue;
    }

    acceptable_order_context.emplace_back(oc);
  }

  OrderExecution order_exe;
  PrepareCancelBatchOrdersExecution(acceptable_order_context, &order_exe);

  auto convert_to_poids =
      [](const std::deque<const OrderContext*>& oc_deque) -> std::vector<int64_t> {
    std::vector<int64_t> poids;

    for (const auto* oc : oc_deque) {
      poids.emplace_back(oc->proc_order_id());
    }

    return poids;
  };

  int total_success_order_cnt = 0;
  while (!order_exe.cancel_pending_order_deque.empty()) {
    order_exe.cancel_proceed_order_deque.clear();

    auto timestamp = GetCurrentTimestamp();

    if (!rest_client_->HasEnoughRateLimit(timestamp, ActionType::CANCEL)) {
      LOG(ERROR) << "Hit rate limit. CancelOrder failed";

      fail_proc_order_id_map.emplace(
          "CANCEL_API_RATE_LIMIT_ERROR",
          convert_to_poids(order_exe.cancel_pending_order_deque));
      break;
    }

    if (!processor_->order_rate_tracker()->HasEnoughRateLimit(timestamp)) {
      LOG(ERROR) << "Hit rate limit. CancelOrder failed";

      fail_proc_order_id_map.emplace(
          "CANCEL_API_RATE_LIMIT_ERROR",
          convert_to_poids(order_exe.cancel_pending_order_deque));
      break;
    }

    if (http_driver_->is_blocking()) {
      fail_proc_order_id_map.emplace(
          "CONNECTION_POOL_FULL",
          convert_to_poids(order_exe.cancel_pending_order_deque));
      break;
    }

    auto request = rest_client_->CancelOrderNorm(order_exe);
    if (!request) {
      std::vector<int64_t> all_proc_order_ids =
          convert_to_poids(order_exe.cancel_pending_order_deque);
      for (const auto* oc : order_exe.cancel_proceed_order_deque) {
        all_proc_order_ids.emplace_back(oc->proc_order_id());
      }
      fail_proc_order_id_map.emplace("CANCEL_API_LIMIT_ERROR", std::move(all_proc_order_ids));
      break;
    }

    PrepareCancelBatchExecutionProto(convert_to_poids(order_exe.cancel_proceed_order_deque));
    processor_->ProcessRawMessage(result_);  // ORDER_CANCELED

    std::vector<const OrderContext*> oc_list(
        order_exe.cancel_proceed_order_deque.begin(),
        order_exe.cancel_proceed_order_deque.end());
    request->handler =
        std::bind(&BaseOrderGateway::HandleResponse, this, oc_list, std::placeholders::_1);
    http_driver_->PostAsyncContext(request);

    total_success_order_cnt += oc_list.size();
  }

  for (auto& [error, proc_order_ids] : fail_proc_order_id_map) {
    PrepareCancelBatchExecutionProto(proc_order_ids);
    if (error == "ORDER_NOT_FOUND") {
      SetOrderExecutionError(error, coin::proto::OrderErrorCode::ORDER_NOT_FOUND);
    } else {
      if (error == "CANCEL_TIMEOUT") {
        for (const auto& id : proc_order_ids) {
          // add to cancel_timeout list
          result_.order_execution.add_cancel_timeout_proc_order_ids(id);
        }
      }
      SetOrderExecutionError(error);
    }
    processor_->ProcessRawMessage(result_);
    result_.order_execution.mutable_error_code()->Clear();
    if (error == "CANCEL_TIMEOUT") {
      result_.order_execution.clear_cancel_timeout_proc_order_ids();
    }
  }

  return total_success_order_cnt;
}

void BaseOrderGateway::PrepareCancelOrdersByProduct(
    int64_t timestamp,
    const std::string& product,
    OrderExecution::SymbolOrderMap* pending_symbol_order_map) {
  std::vector<const OrderContext*> oc_list;
  for (auto order_context : order_manager().GetAllProductWorkingOrders(product)) {
    if (order_context == nullptr) {
      continue;
    }
    auto* oc = const_cast<OrderContext*>(order_context);
    auto proc_order_id = oc->proc_order_id();
    if (oc->is_cancel_pending()) {
      if (timestamp - oc->cancel_created_time() > 5e9) {
        LOG(INFO) << "[CancelOrderAll] No cancel response for order " << proc_order_id << ", "
                  << oc->exchange_order_id();

        // add to timeout cancel list.
        result_.order_execution.add_cancel_timeout_proc_order_ids(oc->proc_order_id());
      }
      LOG(INFO) << "[CancelOrderAll] Cancel has been sent!" << proc_order_id << ", "
                << oc->exchange_order_id();
      continue;
    }
    if (coin::proto::OrderDuration::IOC_ORDER == oc->order_spec().duration ||
        coin::proto::OrderDuration::FOK_ORDER == oc->order_spec().duration) {
      LOG(INFO) << "[CancelOrderAll] Not allow cancel IOC/FOK order: " << proc_order_id;
      continue;
    }
    if (oc->IsCancelConfirmed() || oc->IsFullyFilled()) {
      LOG(INFO) << "[CancelOrderAll] Ignore cancel dead order: " << proc_order_id
                << " status: " << oc->status();
      continue;
    }
    oc_list.emplace_back(oc);
  }
  if (oc_list.size() > 0) {
    pending_symbol_order_map->emplace(product, oc_list);
  }
}

bool BaseOrderGateway::PrepareCancelAllOrderExecution(
    int64_t timestamp,
    const std::string& symbol,
    OrderExecution* order_exe) {
  CHECK(IsSupportMultiCancelOrder())
      << "[CancelOrderAll] unsupport cancel order all. mea: " << mea_.String();

  if (IsCancelOrderAllByAccount(symbol)) {
    for (const auto& product : order_manager().native_symbols()) {
      PrepareCancelOrdersByProduct(timestamp, product, &order_exe->pending_symbol_order_map);
    }
  } else if (order_manager().IsManagedSymbol(symbol)) {
    PrepareCancelOrdersByProduct(timestamp, symbol, &order_exe->pending_symbol_order_map);
  } else {
    LOG(ERROR) << "[CancelOrderAll] error. unknown symbol: " << symbol;
    return false;
  }
  order_exe->is_cancel_order_all = true;
  return true;
}

void BaseOrderGateway::PrepareSubmitBatchOrdersExecution(
    const std::deque<const OrderSpec*>& order_spec_deque,
    OrderExecution* order_exe) {
  order_exe->is_place_batch_orders = true;
  order_exe->place_pending_order_deque = order_spec_deque;
}

void BaseOrderGateway::PrepareCancelBatchOrdersExecution(
    const std::deque<const OrderContext*>& order_context_deque,
    OrderExecution* order_exe) {
  order_exe->is_cancel_batch_orders = true;
  order_exe->cancel_pending_order_deque = order_context_deque;
}

bool BaseOrderGateway::CancelOrderAll(const std::string& native_symbol) {
  int64_t timestamp = GetCurrentTimestamp();
  ClearResult();
  PrepareCancelAllExecutionProto(native_symbol);

  OrderExecution order_exe;
  if (!PrepareCancelAllOrderExecution(timestamp, native_symbol, &order_exe)) {
    LOG(ERROR) << "[CancelOrderAll] ignored. failed to prepare order execution for symbol: "
               << native_symbol;

    SetOrderExecutionError("CANCEL_API_LIMIT_ERROR");
    processor_->ProcessRawMessage(result_);
    return false;
  }

  if (order_exe.pending_symbol_order_map.empty()) {
    LOG(ERROR) << "[CancelOrderAll] ignored. no order found for symbol: " << native_symbol;

    SetOrderExecutionError("CANCEL_API_LIMIT_ERROR");
    processor_->ProcessRawMessage(result_);
    return false;
  }

  while (!order_exe.pending_symbol_order_map.empty()) {
    order_exe.proceed_symbol_order_map.clear();
    if (mea_.exchange == ExchangeType::Kraken) {
      CHECK(false) << "TODO(daniel): to implement";
    } else if (mea_.exchange == ExchangeType::Bybit) {
      CHECK(false) << "TODO(daniel): to implement";
    } else if (
        !(fix_client_ && fix_client_->IsAvailable()) &&
        !rest_client_->HasEnoughRateLimit(timestamp, ActionType::CANCEL)) {
      LOG(ERROR) << "Hit rate limit. CancelOrder failed";

      SetOrderExecutionError("CANCEL_API_LIMIT_ERROR");
      processor_->ProcessRawMessage(result_);
      return false;
    }
    if (!processor_->order_rate_tracker()->HasEnoughRateLimit(timestamp)) {
      LOG(ERROR) << "[CancelOrderAll] Hit rate limit. symbol" << native_symbol;

      SetOrderExecutionError("CANCEL_API_LIMIT_ERROR");
      processor_->ProcessRawMessage(result_);
      return false;
    }
    if (fix_client_ && fix_client_->IsAvailable()) {
      CHECK(false) << "TODO(daniel): to implement";
      fix_client_->CancelOrderAll(order_exe);
      // cancel submit
      std::vector<const OrderContext*> oc_list;
      for (const auto& symbol_order : order_exe.proceed_symbol_order_map) {
        oc_list.insert(oc_list.end(), symbol_order.second.begin(), symbol_order.second.end());
      }

      processor_->ProcessRawMessage(result_);
    } else {
      if (http_driver_->is_blocking()) {
        SetOrderExecutionError("CONNECTION_POOL_FULL");
        processor_->ProcessRawMessage(result_);
        return false;
      }

      std::shared_ptr<AsioHttpContext> res = rest_client_->CancelOrderNorm(order_exe);
      if (!res) {
        SetOrderExecutionError("CANCEL_API_LIMIT_ERROR");
        processor_->ProcessRawMessage(result_);
        return false;
      }
      std::vector<const OrderContext*> oc_list;
      for (const auto& symbol_order : order_exe.proceed_symbol_order_map) {
        oc_list.insert(oc_list.end(), symbol_order.second.begin(), symbol_order.second.end());
      }
      // cancel submit
      processor_->ProcessRawMessage(result_);

      res->handler = std::bind(
          &BaseOrderGateway::HandleResponse,
          this,
          std::move(oc_list),
          std::placeholders::_1);
      http_driver_->PostAsyncContext(res);
    }

    // 20210713(daniel): disable log latency for multi cancel first
    // record latency
    // if (latency_recorder_) {
    //   for (const auto& sybmol_order : order_exe.proceed_symbol_order_map) {
    //     for (const auto& oc : sybmol_order.second) {
    //       latency_recorder_->SetBeforeCancelMark(oc, order_logger_->request());
    //     }
    //   }
    // }
  }
  return true;
}

auto BaseOrderGateway::SubmitBatchOrders(const BatchOrdersSpec& batch_orders_spec) -> int {
  CHECK(IsSupportMultiSubmitOrder())
      << "[" << __func__ << "] unsupport submit batch order. mea: " << mea_.String();

  ClearResult();

  if (batch_orders_spec.Empty()) {
    LOG(ERROR) << "[" << __func__ << "] ignored. no order need submit";
    return 0;
  }
  auto order_spec_deque = batch_orders_spec.GetOrderSpecs();  // need a copy

  using FailErrorType = std::string;
  using OrderSpecList = std::deque<const OrderSpec*>;

  std::unordered_map<FailErrorType, OrderSpecList> fail_order_spec_set;

  OrderSpecList pending_left_order_spec_list;
  OrderSpecList unacceptable_order_spec_list;
  for (const auto& order_spec : order_spec_deque) {
    order_spec.proc_order_id = mutable_order_manager()->AllocateProcOrderIdAndInc();
    // deal with unsupport IOC exchanges
    if ((order_spec.duration == coin::proto::OrderDuration::FOK_ORDER ||
         order_spec.duration == coin::proto::OrderDuration::IOC_ORDER) &&
        !is_supporting_ioc_) {
      order_spec.ResetDuration();
    }

    if (!IsAcceptableValue(order_spec)) {
      unacceptable_order_spec_list.emplace_back(&order_spec);
      continue;
    }

    pending_left_order_spec_list.emplace_back(&order_spec);
  }
  if (!unacceptable_order_spec_list.empty()) {
    fail_order_spec_set.emplace("ORDER_SPEC_NOT_ACCEPTED", std::move(unacceptable_order_spec_list));
  }

  OrderExecution order_exe;
  PrepareSubmitBatchOrdersExecution(pending_left_order_spec_list, &order_exe);

  int total_success_order_cnt = 0;
  while (!order_exe.place_pending_order_deque.empty()) {
    order_exe.place_proceed_order_deque.clear();

    auto timestamp = GetCurrentTimestamp();
    if (!rest_client_->HasEnoughRateLimit(timestamp, ActionType::PLACE)) {
      LOG(ERROR) << "[" << __func__ << "] Hit rate limit. SubmitBatchOrders failed";

      fail_order_spec_set.emplace(
          "SUBMIT_API_RATE_LIMIT_ERROR",
          std::move(order_exe.place_pending_order_deque));
      break;
    }
    if (!processor_->order_rate_tracker()->HasEnoughRateLimit(timestamp)) {
      LOG(ERROR) << "[" << __func__ << "] Hit rate limit. SubmitBatchOrders failed";

      fail_order_spec_set.emplace(
          "SUBMIT_API_LIMIT_ERROR",
          std::move(order_exe.place_pending_order_deque));
      break;
    }

    if (http_driver_->is_blocking()) {
      fail_order_spec_set.emplace(
          "CONNECTION_POOL_FULL",
          std::move(order_exe.place_pending_order_deque));
      break;
    }

    auto context = rest_client_->PlaceOrderNorm(order_exe);
    if (!context) {
      auto all_order_deque = std::move(order_exe.place_pending_order_deque);
      for (const auto* order_spec : order_exe.place_proceed_order_deque) {
        all_order_deque.emplace_back(order_spec);
      }
      fail_order_spec_set.emplace("SUBMIT_API_LIMIT_ERROR", std::move(all_order_deque));
      break;
    }

    PrepareSubmitBatchExecutionProto(order_exe.place_proceed_order_deque);
    processor_->ProcessRawMessage(result_);  // ORDER_SUBMITTED

    std::vector<const OrderContext*> oc_list;
    for (const auto* order_spec : order_exe.place_proceed_order_deque) {
      auto* oc = order_manager().FindByProcOrderId(order_spec->proc_order_id);
      CHECK(oc);
      oc_list.emplace_back(oc);
    }

    context->handler =
        std::bind(&BaseOrderGateway::HandleResponse, this, oc_list, std::placeholders::_1);
    http_driver_->PostAsyncContext(context);

    for (const auto* oc : oc_list) {
      mutable_order_manager()->PostValidateOrderContext(oc);
    }
    total_success_order_cnt += oc_list.size();
  }

  for (auto& [error, order_spec_list] : fail_order_spec_set) {
    PrepareSubmitBatchExecutionProto(order_spec_list);
    SetOrderExecutionError(error);
    processor_->ProcessRawMessage(result_);
    result_.order_execution.mutable_error_code()->Clear();
  }

  return total_success_order_cnt;
}

std::optional<int64_t>
BaseOrderGateway::AmendOrder(int64_t orig_proc_order_id, double amend_price, double amend_qty) {
  int64_t latency_context_id = 0;
#ifdef ENABLE_LATENCY_PROFILE_DEBUG_MARK
  latency_context_id = GetCurrContextIdWithNoRegisteredOG();
  LatencyRecorderGlobal::SetLatencyProfileMark(
      order_logger_->request(),
      "BEFORE_OG_ORDER_AMEND",
      0,
      &latency_context_id,
      false);
#endif

  ClearResult();

  // check order spec/state that if amend order is allowed or not
  auto* orig_oc = order_manager().FindByProcOrderId(orig_proc_order_id);
  if (orig_oc == nullptr) {
    // unexpected. should investigate why a non-exist order is trying be amended
    LOG(ERROR) << "[AmendOrder] original order not found! " << orig_proc_order_id;
    LOG(INFO) << order_manager().DebugString();

    SetOrderExecutionError("ORIGIN_ORDER_NOT_FOUND", coin::proto::OrderErrorCode::ORDER_NOT_FOUND);
    processor_->ProcessRawMessage(result_);
    return {};
  }

  const auto& orig_order_spec = orig_oc->order_spec();
  auto amend_order_spec =
      orig_order_spec.CloneAsAmendOrder(orig_proc_order_id, amend_price, amend_qty);
  int64_t proc_order_id = order_manager().AllocateProcOrderId();
  amend_order_spec.proc_order_id = proc_order_id;
  PrepareAmendOrderExecutionProto(
      proc_order_id,
      orig_proc_order_id,
      amend_price,
      amend_qty,
      amend_order_spec);
  auto timestamp = result_.timestamp;

  if (!(amend_order_spec.is_price_amended || amend_order_spec.is_qty_amended)) {
    // ignored. nothing is amended to the original order
    LOG(ERROR) << "[AmendOrder] amend order with no price/qty changed is ignored: " << proc_order_id
               << ", amend_price=" << amend_price << ", amend_qty=" << amend_qty;
    SetOrderExecutionError("NO_PRICE_QTY_CHANGE");
    processor_->ProcessRawMessage(result_);
    return {};
  }
  if (orig_oc->is_cancel_pending()) {
    // ignored.
    LOG(ERROR) << "[AmendOrder] amend a pending cancel order is not allowed. " << orig_proc_order_id
               << ", " << orig_oc->exchange_order_id();
    SetOrderExecutionError("TARGET_ORDER_IS_CANCEL_PENDING");
    processor_->ProcessRawMessage(result_);
    return {};
  }
  if (coin::proto::OrderDuration::IOC_ORDER == orig_oc->order_spec().duration ||
      coin::proto::OrderDuration::FOK_ORDER == orig_oc->order_spec().duration) {
    // ignored.
    LOG(ERROR) << "[AmendOrder] amend IOC/FOK order is not allowed: " << orig_proc_order_id;
    SetOrderExecutionError("NO_AMEND_FOR_IOC_FOK_ORDER");
    processor_->ProcessRawMessage(result_);
    return {};
  }
  if (orig_oc->IsCancelConfirmed() || orig_oc->IsFullyFilled()) {
    // unexpected. should investigate why a dead order is trying to be amended
    LOG(ERROR) << "[AmendOrder] amend a dead order is ignored: " << orig_proc_order_id
               << " status: " << orig_oc->status();

    SetOrderExecutionError("ORIG_ORDER_IS_DEAD");
    processor_->ProcessRawMessage(result_);
    return {};
  }
  // api rate limit checking. TODO(daniel): can we leave this checking to native client?
  if (orig_oc->order_spec().exchange == ExchangeType::Kraken) {
    auto rest_client = dynamic_cast<KrakenNativePrivateClient*>(rest_client_);
    if (!rest_client->HasEnoughRateLimit(timestamp, ActionType::AMEND, orig_oc->native_product())) {
      LOG(ERROR) << "[AmendOrder] Hit rate limit. AmendOrder failed";

      SetOrderExecutionError("AMEND_API_LIMIT_ERROR");
      processor_->ProcessRawMessage(result_);
      return {};
    }
  } else if (orig_oc->order_spec().exchange == ExchangeType::Bybit) {
    // leave this checking to native client
  } else if (
      !(fix_client_ && fix_client_->IsAvailable()) &&
      !rest_client_->HasEnoughRateLimit(timestamp, ActionType::AMEND)) {
    LOG(ERROR) << "[AmendOrder] Hit rate limit. AmendOrder failed";

    SetOrderExecutionError("AMEND_API_LIMIT_ERROR");
    processor_->ProcessRawMessage(result_);
    return {};
  }
  if (!processor_->order_rate_tracker()->HasEnoughRateLimit(timestamp)) {
    LOG(ERROR) << "[AmendOrder] Hit rate limit. AmendOrder failed";

    SetOrderExecutionError("AMEND_API_LIMIT_ERROR");
    processor_->ProcessRawMessage(result_);
    return {};
  }

  OrderExecution order_exe;
  order_exe.order_spec = &amend_order_spec;
  order_exe.proc_order_id = amend_order_spec.proc_order_id;
  order_exe.oc = const_cast<OrderContext*>(orig_oc);

  if (fix_client_ && fix_client_->IsAvailable()) {
    processor_->ProcessRawMessage(result_);
    fix_client_->AmendOrder(order_exe);
  } else {
    if (http_driver_->is_blocking()) {
      SetOrderExecutionError("CONNECTION_POOL_FULL");
      processor_->ProcessRawMessage(result_);
      return {};
    }

    auto request = rest_client_->AmendOrderNorm(order_exe);
    if (!request) {
      SetOrderExecutionError("AMEND_API_LIMIT_ERROR");
      processor_->ProcessRawMessage(result_);
      return {};
    }
    processor_->ProcessRawMessage(result_);
    auto new_oc = order_manager().FindByProcOrderId(amend_order_spec.proc_order_id);
    std::vector<const OrderContext*> oc_list;
    oc_list.emplace_back(new_oc);
    request->context_id = latency_context_id;
    request->handler =
        std::bind(&BaseOrderGateway::HandleResponse, this, oc_list, std::placeholders::_1);
    http_driver_->PostAsyncContext(request);
    mutable_order_manager()->PostValidateOrderContext(new_oc);
  }

  if (latency_recorder_) {
    auto* amend_oc = order_manager().FindByProcOrderId(amend_order_spec.proc_order_id);
    latency_recorder_->SetBeforeAmendMark(amend_oc, order_logger_->request());
  }

  return amend_order_spec.proc_order_id;
}

bool BaseOrderGateway::SubmitConversion(const ConversionSpec& conv_spec) {
  ClearResult();
  if (fix_client_ && fix_client_->IsAvailable()) {
    CHECK(false) << "TODO(daniel): do not support submit conversion via fix client yet";
  } else {
    if (http_driver_->is_blocking()) {
      return false;
    }
    auto request = rest_client_->SubmitConversion(conv_spec);
    if (!request) {
      return false;
    }
    std::vector<const OrderContext*> dummy_oc_list;
    request->handler =
        std::bind(&BaseOrderGateway::HandleResponse, this, dummy_oc_list, std::placeholders::_1);
    http_driver_->PostAsyncContext(request);
  }

  return true;
}

bool BaseOrderGateway::SubmitLiquidity(const LiquiditySpec& liquidity_spec) {
  ClearResult();
  if (grpc_client_ == nullptr) {
    CHECK(false) << "SubmitLiquidity only support grpc_client!";
  }

  grpc_client_->SubmitLiquidity(liquidity_spec);

  return true;
}

void BaseOrderGateway::GetRateLimitInfo(const std::shared_ptr<AsioHttpContext>& http_context) {
  if (symcache_->mea().exchange != ExchangeType::Bitmex) {
    return;
  }

  int32_t ratelimit = -1;
  int32_t ratelimit_remain = -1;
  for (auto&& header : http_context->res.base()) {
    if (header.name_string() == "X-RateLimit-Remaining") {
      ratelimit_remain = std::stoi(std::string(header.value()));
    } else if (header.name_string() == "X-RateLimit-Limit") {
      ratelimit = std::stoi(std::string(header.value()));
    }
    if (ratelimit >= 0 && ratelimit_remain >= 0) {
      rest_client_->rate_tracker()->SetRateLimitInfo(ratelimit, ratelimit_remain);
      LOG_EVERY_N(INFO, 10) << "ratelimit: " << ratelimit << ", " << ratelimit_remain;
      return;
    }
  }
}

void BaseOrderGateway::set_latency_recorder(
    coin2::exchange::base::strategy_util::LatencyRecorder* latency_recorder) {
  latency_recorder_ = latency_recorder;
  processor_->latency_recorder_ = latency_recorder;
  if (http_driver_ != nullptr) {
    http_driver_->set_latency_recorder(latency_recorder);
  }
}

auto BaseOrderGateway::GetCurrContextIdWithNoRegisteredOG() -> int64_t {
  // TODO(daniel): workaround first
  if (!latency_recorder_) {
    return 0;
  }
  auto context_id = latency_recorder_->GetCurrentContextId();
  if (og_registered_context_id_set_.count(context_id)) {
    return 0;
  }
  og_registered_context_id_set_.emplace(context_id);
  return context_id;
}

void BaseOrderGateway::RemoveContextIdFromRegisteredOGSet(int64_t context_id) {
  if (0 == og_registered_context_id_set_.count(context_id)) {
    return;
  }
  og_registered_context_id_set_.erase(context_id);
}

BaseOrderSubsystem::BaseOrderSubsystem(
    const std::string& system_name,
    const MarketExchangeApi& mea,
    const di::ConstantInjection& ci,
    const OrderSystemConfig& oss_config,
    executor::LiveAsyncExecutor* exec,
    IOrderSubscriber* callback,
    GetProductHolderFunc get_product_holder)
    : callback_(callback),
      system_name_(system_name),
      mea_(mea),
      oss_config_(oss_config),
      live_executor_(exec),
      get_product_holder_(get_product_holder) {
  live_executor_->RegisterSubsystem(this);
  message_bus_ = std::make_unique<MessageBus>();
  int64_t submit_pending_timeout_sec = oss_config.order_gateway_config().submit_pending_timeout_sec();
  ocm_ = std::make_unique<OrderContextManager>(mea);
  ocm_->SetSubmitPendingTimeout(submit_pending_timeout_sec);

  std::string key_filepath = oss_config.connection_config().key_filepath();
  std::string owner = AuthKey::FromFile(key_filepath).owner();

  account_request_.set_market_type(base::market::MarketTypeToString(mea.market));
  account_request_.set_exchange(base::market::ExchangeTypeToString(mea.exchange));
  account_request_.set_owner(owner);
  account_request_.set_api_version(mea.api);
  account_request_.set_system_name(system_name_);

  auto& order_logger_config = oss_config.order_logger_config();

  if (order_logger_config.as_account_logger()) {
    logger_ = std::make_unique<base::order_util::OrderLogger>(
        account_request_,
        order_logger_config.account_log_root(),
        order_logger_config.machine(),
        order_logger_config.log_extension(),
        order_logger_config.use_slim_log_mode());
    logger_->set_as_account_logger(order_logger_config.as_account_logger());
  } else {
    logger_ = std::make_unique<base::order_util::OrderLogger>(
        account_request_,
        order_logger_config.log_root(),
        order_logger_config.machine(),
        order_logger_config.log_extension(),
        order_logger_config.use_slim_log_mode());
  }
  system_critical_error_callback_ = [logger = logger_.get()](
                                        const char* file,
                                        int line,
                                        std::string_view condition,
                                        std::string_view message) {
    auto error_message = fmt::format("System Critical Error: {} {}", condition, message);

    LOG(ERROR) << error_message;

    auto ts = GetCurrentTimestamp();
    auto time_str = Iso8601FromTimestamp(ts);
    auto pid = getpid();

    auto system_critical_error_message =
        fmt::format("{} {} {}:{}] {}", time_str, pid, file, line, error_message);

    using RiskEvent = coin::proto::RiskEvent;
    RiskEvent risk_event;

    risk_event.set_system_critical_error_message(system_critical_error_message);

    risk_event.set_risk_event_type(RiskEvent::SYSTEM_CRITICAL_ERROR);

    logger->WriteRiskEvent(risk_event, true, ts);

    std::abort();
  };

  CHECK(SystemCriticalError::Get().RegisterCallback(&system_critical_error_callback_))
      << "register fail, please check code!";

  bool enable_raw_og_log = false;
  // temporary workaround until we enable all raw og log
  if (oss_config.order_gateway_config().has_log_private_exchange_message()) {
    enable_raw_og_log = oss_config.order_gateway_config().log_private_exchange_message();
  } else {
    enable_raw_og_log = IsRawOgLogEnabled(mea);
  }

  order_packet_logger_ = std::make_unique<base::order_util::OrderLogger>(
      account_request_,
      order_logger_config.raw_log_root(),
      order_logger_config.machine(),
      order_logger_config.raw_log_extension());

  using NativeHttpAsioDriver = base::api_base::NativeHttpAsioDriver;
  http_driver_ =
      std::make_unique<NativeHttpAsioDriver>(exec->MainIoc(), mea, oss_config.connection_config());
  processor_ = std::make_unique<BaseOrderProcessor>(
      system_name,
      mea,
      callback_,
      logger_.get(),
      order_packet_logger_.get(),
      enable_raw_og_log);
  processor_->set_store_parsed_message(order_logger_config.store_parsed_message());

  logger_->set_og_id(processor_->og_id());
  order_packet_logger_->set_og_id(processor_->og_id());

  ws_subscriber_ = std::make_unique<WsSubscriber>(mea, exec, oss_config.connection_config());
  grpc_client_ = GrpcTradeClientMgr::Create(mea, oss_config.connection_config());
  using FixTradeClientMgr = coin2::exchange::base::order::FixTradeClientMgr;
  fix_client_ = FixTradeClientMgr::Create(mea, oss_config.connection_config());
}

BaseOrderSubsystem::~BaseOrderSubsystem() {
  if (system_critical_error_callback_) {
    CHECK(SystemCriticalError::Get().UnregisterCallback(&system_critical_error_callback_))
        << "unregister fail, please check code!";
  }
}

void BaseOrderSubsystem::LiveInit(
    const OrderSystemConfig& oss_config,
    executor::LiveAsyncExecutor* exec) {
  bool portfolio_margin_trading = oss_config.portfolio_margin_trading();
  bool is_readonly = oss_config.order_logger_config().as_account_logger();
  gateway_ = std::make_unique<BaseOrderGateway>(
      mea_,
      oss_config.order_gateway_config(),
      exec,
      rest_client_.get(),
      http_driver_.get(),
      logger_.get());

  symcache_.set_mea(mea_);
  symcache_.SetGetProductHolder(get_product_holder_);
  symcache_.SetPortfolioMarginTrading(portfolio_margin_trading);
  auto products = ProductRequestsFromConfig(oss_config_.mea(), oss_config_.products());
  for (auto& prod : products) {
    products_.push_back(prod.get());
    symcache_.AddProduct(std::move(prod));
  }

  ocm_->SetSymcache(&symcache_);
  gateway_->SetSymcache(&symcache_);
  processor_->SetSymcache(&symcache_);
  CHECK(rest_subscriber_);
  processor_->SetRestSubscriber(rest_subscriber_.get());
  rest_client_->SetGetProductHolder(get_product_holder_);

  rest_client_->SetMessageBus(message_bus_.get());
  rest_client_->SetSymcache(&symcache_);
  rest_client_->SetOrderLoggerCB(std::bind(
      &order_util::OrderLogger::WriteRateLimitReportMessage,
      logger_.get(),
      std::placeholders::_1,
      std::placeholders::_2,
      std::placeholders::_3));
  rest_client_->Prepare();

  rest_subscriber_->SetSymcache(&symcache_);
  rest_subscriber_->SetProcessor(processor_.get());
  rest_subscriber_->SetFlowSub(live_executor_);
  rest_subscriber_->SetMessageBus(message_bus_.get());
  rest_subscriber_->SetIsReadonlyOrderSystem(is_readonly);
  rest_subscriber_->SetGrpcClient(grpc_client_.get());

  ocm_->RegisterOrderEventHandler(std::bind(
      &BaseOrderProcessor::HandleOrderEvent,
      dynamic_cast<BaseOrderProcessor*>(processor_.get()),
      std::placeholders::_1));
  ocm_->RegisterRiskEventHandler(std::bind(
      &BaseOrderProcessor::HandleRiskEvent,
      dynamic_cast<BaseOrderProcessor*>(processor_.get()),
      std::placeholders::_1));
  processor_->mutable_upd_mgr()->SetOrderManager(ocm_.get());
  logger_->SetCallback(
      std::bind(&base::order::IOrderSubscriber::onOrderLog, callback_, std::placeholders::_1));

  for (auto* prod : symcache_.GetProducts()) {
    ws_subscriber_->RequestNativeSymbol(prod->order_native_symbol(), '\0');
    dynamic_cast<BaseOrderProcessor*>(processor_.get())
        ->AddNativeSymbol(prod->order_native_symbol(), callback_);
  }
  ws_subscriber_->SetProcessor(processor_.get());
  ws_subscriber_->SetMessageBus(message_bus_.get());
  ws_subscriber_->SetIsReadonlyOrderSystem(is_readonly);

  // Remove later.
  message_bus_->Subscribe(
      "upd_ws_last_recv_msg_ts",
      std::bind(
          &WsSubscriber::SetLastRecvMsgTs,
          ws_subscriber_.get(),
          std::placeholders::_1,
          std::placeholders::_2));
  message_bus_->Subscribe(
      "force_reset_wss_connection",
      std::bind(
          &WsSubscriber::ForceResetConnection,
          ws_subscriber_.get(),
          std::placeholders::_1,
          std::placeholders::_2));
  if (mea_.exchange == ExchangeType::Binance) {
    message_bus_->Subscribe(
        "binance/listen_key",
        std::bind(
            &WsSubscriber::SetListenKey,
            ws_subscriber_.get(),
            std::placeholders::_1,
            std::placeholders::_2));
  }
  if (mea_.exchange == ExchangeType::Lbank) {
    message_bus_->Subscribe(
        "lbank/listen_key",
        std::bind(
            &WsSubscriber::SetListenKey,
            ws_subscriber_.get(),
            std::placeholders::_1,
            std::placeholders::_2));
  }
  if (mea_.exchange == ExchangeType::Bit) {
    message_bus_->Subscribe(
        "bit/listen_key",
        std::bind(
            &WsSubscriber::SetListenKey,
            ws_subscriber_.get(),
            std::placeholders::_1,
            std::placeholders::_2));
  }
  if (mea_.exchange == ExchangeType::Mexc) {
    message_bus_->Subscribe(
        "mexc/listen_key",
        std::bind(
            &WsSubscriber::SetListenKey,
            ws_subscriber_.get(),
            std::placeholders::_1,
            std::placeholders::_2));
  }
  if (mea_.exchange == ExchangeType::Kraken) {
    message_bus_->Subscribe(
        "kraken/ws_token",
        std::bind(
            &WsSubscriber::SetListenKey,
            ws_subscriber_.get(),
            std::placeholders::_1,
            std::placeholders::_2));
  }
  if (mea_.exchange == ExchangeType::Gateio) {
    if (mea_.market == MarketType::Futures) {
      message_bus_->Subscribe(
          "gateio_futures/user_id",
          std::bind(
              &WsSubscriber::SetUserId,
              ws_subscriber_.get(),
              std::placeholders::_1,
              std::placeholders::_2));
    }
  }
  if (mea_.exchange == ExchangeType::Kucoin) {
    message_bus_->Subscribe(
        "kucoin/bullet",
        std::bind(
            &WsSubscriber::SetBullet,
            ws_subscriber_.get(),
            std::placeholders::_1,
            std::placeholders::_2));
  }
  if (mea_.exchange == ExchangeType::Apollox) {
    message_bus_->Subscribe(
        "apollox/listen_key",
        std::bind(
            &WsSubscriber::SetListenKey,
            ws_subscriber_.get(),
            std::placeholders::_1,
            std::placeholders::_2));
  }
  if (mea_.exchange == ExchangeType::Bittrex) {
    message_bus_->Subscribe(
        "bittrex/listen_key",
        std::bind(
            &WsSubscriber::SetListenKey,
            ws_subscriber_.get(),
            std::placeholders::_1,
            std::placeholders::_2));
  }

  processor_->SetOcm(ocm_.get());
  processor_->SetParser(parser_.get());
  processor_->SetMessageBus(message_bus_.get());
  processor_->Prepare();

  parser_->SetOcm(ocm_.get());
  parser_->SetSymcache(&symcache_);
  parser_->SetMessageBus(message_bus_.get());
  parser_->SetIsReadonlyOrderSystem(is_readonly);

  gateway_->SetProcessor(processor_.get());
  gateway_->SetOcm(ocm_.get());
  gateway_->SetMessageBus(message_bus_.get());
  gateway_->Prepare();

  if (fix_client_) {
    fix_client_->SetExecutor(live_executor_->MainIoc());
#ifndef DISABLE_REACTOR_INTEGRATION
    fix_client_->SetMainReactor(live_executor_->main_reactor());
#endif
    fix_client_->SetProcessor(processor_.get());
    fix_client_->SetGetProductHolder(get_product_holder_);
    gateway_->SetFixClient(fix_client_.get());
  }
  if (grpc_client_) {
    grpc_client_->SetCallback(
        std::bind(&BaseOrderGateway::HandleGrpcResponse, gateway_.get(), std::placeholders::_1));
    gateway_->SetGrpcClient(grpc_client_.get());
  }
  SetOsFeature(mea_);
}

void BaseOrderSubsystem::StartAsync() {
  ws_subscriber_->RegisterPubTopic();
  rest_subscriber_->RegisterPubTopic();

  ws_subscriber_->StartAsync();
  rest_subscriber_->StartAsync();

  http_driver_->Start();
  if (fix_client_) {
    fix_client_->Start();
  }
}

void BaseOrderSubsystem::StopAsync() {
  ws_subscriber_->StopAsync();
  rest_subscriber_->StopAsync();
  http_driver_->Stop();
  if (fix_client_) {
    fix_client_->Stop();
  }
}

bool BaseOrderSubsystem::is_supporting_separate_pos() const {
  return os_feature_.is_supporting_separate_pos();
}

void BaseOrderSubsystem::SetOsFeature(const MarketExchangeApi& mea) {
  std::ifstream ifs("data/coin2/order/og_info.json");
  nlohmann::json jf = nlohmann::json::parse(ifs);
  if (jf.find(mea.String()) != jf.end()) {
    auto data = jf[mea.String()];
    if (data.find("need_exchange_id_to_cancel") != data.end()) {
      os_feature_.set_need_exchange_id_to_cancel(data["need_exchange_id_to_cancel"]);
    }
    if (data.find("is_supporting_separate_pos") != data.end()) {
      os_feature_.set_is_supporting_separate_pos(data["is_supporting_separate_pos"]);
    }
    if (data.find("is_supporting_ioc") != data.end()) {
      os_feature_.set_is_supporting_ioc(data["is_supporting_ioc"]);
    }
    if (data.find("is_supporting_post_only") != data.end()) {
      os_feature_.set_is_supporting_post_only(data["is_supporting_post_only"]);
    }
  }
}

bool BaseOrderSubsystem::is_ready() const {
  if (!account_is_ready_) {
    account_is_ready_ = processor_->upd_mgr().account_info().is_ready();
  }
  return account_is_ready_;
}

bool RestSubscriber::SkipQuery(const std::string& task_name, const std::string& native_symbol) {
  // has active order
  auto& live_orders = processor_->order_manager()->GetAllProductWorkingOrders(native_symbol);
  if (live_orders.size() > 0) {
    return false;
  }
  // has order event recently (in 5 mins)
  if (processor_->HasActiveEvent(native_symbol)) {
    return false;
  }
  // TODO(daniel): pass for first time?

  return true;
}

void RestSubscriber::HandleQuery(const std::shared_ptr<AsioHttpContext>& http_context) {
  if (!http_context) {
    return;
  }
  if (rest_client_ && rest_client_->rate_tracker()) {
    rest_client_->rate_tracker()->ParseRateLimitInfo(http_context, ActionType::QUERY);
  }

  switch (http_context->status) {
    case AsioHttpContextStatus::kResponded: {
      BaseParser::RawMessage msg;
      msg.type = OrderSubTopicId::REST_MESSAGE;
      msg.http_context = http_context;
      msg.timestamp = http_context->response_timestamp;
      processor_->ProcessRawMessage(msg);
      break;
    }
    case AsioHttpContextStatus::kTimeout:  // for timedout query, the data may be outdated,
    case AsioHttpContextStatus::kError: {  // so just ignore it
      // update only 1. message id 2. timestamp
      BaseParser::RawMessage msg;
      msg.type = OrderSubTopicId::REST_MESSAGE;
      msg.timestamp = http_context->response_timestamp;
      processor_->PreProcessMessage(msg);
      LOG(INFO) << "connection error!";
      break;
    }
    case AsioHttpContextStatus::kRequested: {
      break;
    }
    default: {
      CHECK(false) << "http context status: " << static_cast<int>(http_context->status);
      break;
    }
  }

  // write rest query error (read/write/timeout etc)
  processor_->HandleHttpError(http_context);
}

void BaseOrderSubsystem::GetApiEndPointStats(RequestStatus info) {
  if (!reporter_func_) {
    return;
  }
  reporter_func_(info);
}

// Exported methods
std::map<std::string, std::function<std::string(const std::string&)>>
BaseOrderSubsystem::ExportMethods() {
  return {
      {"get_oss_features", [this](const std::string& req) { return GetOssFeatures(req); }},
      {"get_oss_configure", [this](const std::string& req) { return GetOssConfig(req); }},
      {"get_open_orders", [this](const std::string& req) { return GetOpenOrders(req); }},
      {"set_raw_og_log", [this](const std::string& req) { return SetRawOGLog(req); }},
      {"get_og_status", [this](const std::string& req) { return GetOGInfo(req); }},
      {"get_git_revision", [this](const std::string& req) { return GetGitHash(req); }},
      {"get_raw_order_data", [this](const std::string& req) { return GetRawOrderData(req); }}};
}

std::string BaseOrderSubsystem::GetOGInfo(const std::string&) {
  auto info = processor_->GetOGInfo();
  auto order_info = info.first;
  nlohmann::json j = nlohmann::json{
      {"order_reject_num", order_info.order_reject_num},
      {"order_accept_num", order_info.order_accept_num},
      {"cancel_reject_num", order_info.cancel_reject_num},
      {"ws_reset_num", info.second}};
  return j.dump(4);
}

std::string BaseOrderSubsystem::GetOssFeatures(const std::string&) {
  return ProtoToJsonString(os_feature_);
}

std::string BaseOrderSubsystem::SetRawOGLog(const std::string& req) {
  std::string rep = "";
  if (req == "false") {
    SetRawOGLog(false);
    rep = "Set Raw OG Log: false";
  } else {
    SetRawOGLog(true);
    rep = "Set Raw OG Log: true";
  }
  return rep;
}

std::string BaseOrderSubsystem::GetOssConfig(const std::string& req) {
  return ProtoToJsonString(oss_config_);
}

std::string BaseOrderSubsystem::GetGitHash(const std::string& req) {
  auto build_info = GetBuildInfo();

  return "git revision: " + build_info.revision();
}

std::string BaseOrderSubsystem::GetOpenOrders(const std::string& req) {
  nlohmann::json json;
  auto oc_list = ocm_->GetAllWorkingOrders();
  for (auto oc : oc_list) {
    json.push_back(ProtoToJsonString(oc->ToProto()));
  }
  return json.dump(4);
}

std::string BaseOrderSubsystem::GetRawOrderData(const std::string& req) {
  nlohmann::json json;
  auto [ts, data] = processor_->GetRawOrderData();
  json["timestamp"] = ts;
  json["data"] = data;
  return json.dump(4);
}

}  // namespace coin2::exchange::base::order
